package log4go

import (
	"encoding/json"
	"fmt"
	"os"
	"time"

	zmq "github.com/pebbe/zmq4"
)

const (
	CONST_DTIME     = 2
	CONST_LCOUNT    = 50
	CONST_SCANSIZE  =            10
	CONST_SCAN_TIME =            20
)

type zmqLogRecord struct {
	Target  string
	Level   string    // The log level
	Created time.Time // The time at which the log message was created (nanoseconds)
	Source  string    // The message source
	Message string    // The log message
}

// This log writer sends output to a socket
type ZmqLogWriter chan *LogRecord

// This is the SocketLogWriter's output method
func (w ZmqLogWriter) LogWrite(rec *LogRecord) {
	w <- rec
}

func (w ZmqLogWriter) Close() {
	close(w)
}

type SourceInfo struct {
	Level   Level
	Created time.Time
	Count   int
}

//log control
type FlowControl struct {
	DTime   int64
	LCount  int
	FlowMap map[string]*SourceInfo
}

func (this *FlowControl) Init(dTime int64, limit int) {
	this.DTime = dTime
	this.LCount = limit
	this.FlowMap = make(map[string]*SourceInfo, 0)
}

//timeout sacn
func (this *FlowControl) Scan(maxScanSize int) []*LogRecord {
	lrArr := make([]*LogRecord, 0)
	scanNum := 0
	for key, value := range this.FlowMap {
		if (time.Now().Unix()-value.Created.Unix() >= this.DTime) && (value.Count > this.LCount) {
			plr := &LogRecord{value.Level, value.Created, key, fmt.Sprintf("large logs in %ds,throw %d", this.DTime, value.Count-this.LCount)}
			value.Count = 0
			value.Created = time.Now()
			lrArr = append(lrArr, plr)
			scanNum++
			if (scanNum) >= maxScanSize {
				break
			}
		}
	}
	return lrArr
}

//monitor every log
func (this *FlowControl) Monitor(plr *LogRecord) (*LogRecord, bool) {
	value, ok := this.FlowMap[plr.Source]
	if !ok {
		this.FlowMap[plr.Source] = &SourceInfo{plr.Level, plr.Created, 1}
		return plr, true
	}

	if plr.Created.Unix()-value.Created.Unix() < this.DTime {
		value.Count++
		if value.Count > this.LCount {
			return nil, false
		}
		return plr, true
	} else {
		if value.Count > this.LCount {
			newPlr := &LogRecord{plr.Level,plr.Created,plr.Source,fmt.Sprintf("large logs in %ds,throw %d", this.DTime, value.Count-this.LCount)}
//			plr.Message = fmt.Sprintf("large logs in %ds,throw %d", this.DTime, value.Count-this.LCount)
			plr = newPlr
		}
		value.Count = 0
		value.Created = plr.Created
		return plr, true
	}
}

func NewZmqLogWriter(hostport, target, topic string) ZmqLogWriter {
	sock, err := zmq.NewSocket(zmq.PUB)
	//	sock, err := net.Dial(proto, hostport)
	if err != nil {
		fmt.Fprintf(os.Stderr, "NewZmqLogWriter(%q): %s\n", hostport, err)
		return nil
	}

	if serr := sock.SetLinger(0); serr != nil {
		fmt.Fprintf(os.Stderr, "NewZmqLogWriter(%q): %s\n", hostport, serr)
		return nil
	}

	if serr := sock.SetSndhwm(100); serr != nil {
		fmt.Fprintf(os.Stderr, "NewZmqLogWriter(%q): %s\n", hostport, serr)
		return nil
	}

	//	if serr := sock.SetConflate(true); serr != nil {
	//		fmt.Fprintf(os.Stderr, "NewZmqLogWriter(%q): %s\n", hostport, serr)
	//		return nil
	//	}

	if cerr := sock.Connect(hostport); cerr != nil {
		fmt.Fprintf(os.Stderr, "NewZmqLogWriter(%q): %s\n", hostport, cerr)
		return nil
	}

	w := ZmqLogWriter(make(chan *LogRecord, LogBufferLength * 8))

//	logControl := &FlowControl{}
//	logControl.Init(CONST_DTIME, CONST_LCOUNT)

	go func() {
		defer func() {
			if sock != nil {
				sock.Disconnect(hostport)
				sock.Close()
			}
		}()
//		scanTimer := time.NewTicker(time.Second * time.Duration(CONST_SCAN_TIME))
//        defer scanTimer.Stop()
		zlr := &zmqLogRecord{Target: target}
		for {
			plr, ok := <-w
			if !ok {
				return
			}
			zmqSend(plr, zlr, sock, &hostport, &topic)
		}
//		for {
//			select {
//			case <-scanTimer.C:
//				lrArr := logControl.Scan(CONST_SCANSIZE)
//				for i := 0; i < len(lrArr); i++ {
//					zmqSend(lrArr[i], zlr, sock, &hostport, &topic)
//				}
//			case plr, ok := <-w:
//				if !ok {
//					return
//				}
//				if newPlr, sendFlag := logControl.Monitor(plr); sendFlag {
//					zmqSend(newPlr, zlr, sock, &hostport, &topic)
//				}
//			}
//		}
	}()
	return w
}

func zmqSend(rec *LogRecord, zlr *zmqLogRecord, sock *zmq.Socket, hostport, topic *string) {
	// Marshall into JSON
	zlr.Level = levelStrings[rec.Level]
	zlr.Created = rec.Created
	zlr.Source = rec.Source
	zlr.Message = rec.Message
	js, err := json.Marshal(zlr)
	if err != nil {
		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", *hostport, err)
		return
	}
	if _, err = sock.SendBytes(([]byte)(*topic), zmq.SNDMORE); err != nil {
		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", *hostport, err)
	}
	if _, err = sock.SendBytes(js, zmq.DONTWAIT); err != nil {
		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", *hostport, err)
	}
}

//for rec := range w {
//	// Marshall into JSON
//	lr.Level = levelStrings[rec.Level]
//	lr.Created = rec.Created
//	lr.Source = rec.Source
//	lr.Message = rec.Message
//	js, err := json.Marshal(lr)
//	if err != nil {
//		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", hostport, err)
//		continue
//	}
//	if _, err = sock.SendBytes(([]byte)(topic), zmq.SNDMORE); err != nil {
//		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", hostport, err)
//	}

//	if _, err = sock.SendBytes(js, zmq.DONTWAIT); err != nil {
//		fmt.Fprint(os.Stderr, "ZmqLogWriter(%q): %s", hostport, err)
//	}
//}
